import json
import logging
from kafka import KafkaProducer
from kafka.errors import KafkaError
from typing import Optional, Dict, Any

logger = logging.getLogger(__name__)


class KafkaProducerService:
    def __init__(self, bootstrap_servers: str = "localhost:9092"):
        self.bootstrap_servers = bootstrap_servers
        self.producer = None
        self._connect()

    def _connect(self):
        try:
            self.producer = KafkaProducer(
                bootstrap_servers=self.bootstrap_servers,
                value_serializer=lambda v: json.dumps(v).encode('utf-8'),
                key_serializer=lambda k: k.encode('utf-8') if k else None,
                acks='all',
                retries=3,
                retry_backoff_ms=1000
            )
            logger.info(f"Connected to Kafka at {self.bootstrap_servers}")
        except Exception as e:
            logger.error(f"Failed to connect to Kafka: {e}")
            raise

    def send_message(self, topic: str, message: Any, key: Optional[str] = None) -> Dict[str, Any]:
        try:
            future = self.producer.send(
                topic=topic,
                value=message,
                key=key
            )
            
            record_metadata = future.get(timeout=10)
            
            return {
                "success": True,
                "topic": record_metadata.topic,
                "partition": record_metadata.partition,
                "offset": record_metadata.offset
            }
        except KafkaError as e:
            logger.error(f"Failed to send message to Kafka: {e}")
            return {
                "success": False,
                "error": str(e)
            }

    def close(self):
        if self.producer:
            self.producer.close()
            logger.info("Kafka producer closed")